package com.guqj.thread.pool;

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.RejectedExecutionHandler;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.LongAdder;

public class ThreadPool {

	private final static int PRODUCE_TASK_SLEEP_TIME = 2;

	private final static int PRODUCE_TASK_MAX_NUMBER = 10;

	private final static int PRODUCE_TASK_CORE_THREAD_NUMBER = 2;

	private final static int PRODUCE_TASK_MAX_THREAD_NUMBER = 5;

	private final static int PRODUCE_TASK_KEEPALIVE_TIME = 2;

	public void threadPoolTest() {
		ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(PRODUCE_TASK_CORE_THREAD_NUMBER,
				PRODUCE_TASK_MAX_THREAD_NUMBER, PRODUCE_TASK_KEEPALIVE_TIME, TimeUnit.SECONDS,
				new ArrayBlockingQueue<Runnable>(3), new CustomThreadFactory(), new CustomRejectedExecutionHandler());
		threadPoolExecutor.prestartCoreThread();
		for (int i = 1; i <= PRODUCE_TASK_MAX_NUMBER; i++) {
			try {
				String task = "task@ " + i;
				System.out.println("创建任务并提交到线程池中：" + task);
				threadPoolExecutor.execute(new ThreadPoolTask(task));
				Thread.sleep(PRODUCE_TASK_SLEEP_TIME);
			} catch (Exception e) {
				e.printStackTrace();
			}
		}
	}

	private class CustomThreadFactory implements ThreadFactory {
		private LongAdder count = new LongAdder();

		@Override
		public Thread newThread(Runnable r) {
			count.add(1L);
			Thread t = new Thread(r);
			String threadName = ThreadPool.class.getSimpleName() + " " + count.longValue();
			System.out.println(threadName);
			t.setName(threadName);
			return t;
		}
	}

	private class CustomRejectedExecutionHandler implements RejectedExecutionHandler {
		@Override
		public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
			// 记录异常
			// 报警处理等
			System.out.println("error.............");
			try {
				// 核心改造点，由blockingqueue的offer改成put阻塞方法
				executor.getQueue().put(r);
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
		}
	}
}
